import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description:
 * @title: ProdConsumerBlockQueueDemo
 * @author: zp
 * @date: 2021/8/3 20:22
 * 使用消息队列完成多线程操作
 */
class MyResource{
    private AtomicInteger atomicInteger=new AtomicInteger();
    private BlockingQueue<String> blockingQueue;
    private volatile boolean flag=true;

    public MyResource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void add5() throws InterruptedException {
        boolean reValue;
        while (flag){
            //生产消息
            reValue=blockingQueue.offer(String.valueOf(atomicInteger.addAndGet(1)),2L, TimeUnit.SECONDS);
            if(!reValue){
                System.out.println("插入数据失败");

            }
            System.out.println("插入数据成功");
            TimeUnit.SECONDS.sleep(1);
            System.out.println(atomicInteger.get());
        }
        System.out.println("生产者关闭");
    }

    public void ave5() throws InterruptedException {
        String reValue;
        while (flag){
            //消费消息
            reValue=blockingQueue.poll(2L,TimeUnit.SECONDS);
            if(reValue==null || reValue.equalsIgnoreCase("")){
                flag=false;
                System.out.println("消费者关闭");
                return;
            }
            System.out.println("消费数据成功");
            System.out.println(atomicInteger.get());
        }
        System.out.println("生产关闭");
    }

    public void stop(){
        flag=false;
    }
}
/**
 * @description:
 * @author: zp
 * @date: 2021/8/5 20:39
 * @param: null
 * @return:
 */
public class ProdConsumerBlockQueueDemo {
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue=new ArrayBlockingQueue<>(10);
        MyResource myResource=new MyResource(blockingQueue);
        new Thread(()->{
            try {
                myResource.add5();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"AAA").start();

        new Thread(()->{
            try {
                myResource.ave5();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"BBB").start();

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        myResource.stop();

    }
}
